package com.kl.example.admin;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.config.ConfigResource;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class KafkaAdminClientExample {

    private static final String topic = "topic-admin";
    private static final String broker = "localhost:9092";

    private static AdminClient adminClient;

    /**
     * 初始化
     */
    @BeforeClass
    public static void beforeClass() {
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, broker);
        adminClient = AdminClient.create(properties);
    }

    /**
     * 收尾
     */
    @AfterClass
    public static void afterClass() {
        adminClient.close();
    }


    /**
     * 创建一个主题名为 topic-admin，分区数为4，副本因子为1的主题
     */
    @Test
    public void createTopics() {
        NewTopic newTopic = new NewTopic(topic, 4, (short) 1);
        Collection<NewTopic> newTopicList = new ArrayList<>();
        newTopicList.add(newTopic);
        CreateTopicsResult topics = adminClient.createTopics(newTopicList);
        try {
            topics.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列举出所有的Topic
     */
    @Test
    public void listTopics() {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        try {
            KafkaFuture<Collection<TopicListing>> listings = listTopicsResult.listings();
            Collection<TopicListing> topicListings = listings.get();
            for (TopicListing topicListing : topicListings) {
                // (name=topic-admin, internal=false)
                System.out.println(topicListing);
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("==============");
        // 只列举出 topic 的名字
        try {
            KafkaFuture<Set<String>> names = listTopicsResult.names();
            Set<String> strings = names.get();
            for (String topicName : strings) {
                // topic-admin
                System.out.println("topicName=" + topicName);
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 测试和上面的方法结果差不多，不知道 Internal 参数的具体作用
     */
    @Test
    public void listTopicsWithArg() {
        try {
            ListTopicsOptions options = new ListTopicsOptions();
            options.listInternal(true);
            ListTopicsResult listTopicsResult = adminClient.listTopics(options);
            Set<String> strings = listTopicsResult.names().get();
            for (String string : strings) {
                System.out.println(string);
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 描述主题信息
     */
    @Test
    public void describeTopic() {
        Collection<String> topicNames = new ArrayList<>();
        topicNames.add(topic);
        topicNames.add("test");
        DescribeTopicsResult result = adminClient.describeTopics(topicNames);
        try {
            Map<String, TopicDescription> descriptionMap = result.all().get();
            descriptionMap.forEach((tpName, tpDes) -> {
                System.out.println("topicName = " + tpName);
                List<TopicPartitionInfo> partitions = tpDes.partitions();
                System.out.println(tpDes.name() + " 分区信息说明");
                for (TopicPartitionInfo partition : partitions) {
                    System.out.println(partition);
                }
                System.out.println();
                System.out.println();
            });
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除主题
     */
    @Test
    public void deleteTopics() {
        Collection<String> topics = Collections.singletonList(topic);
        DeleteTopicsResult result = adminClient.deleteTopics(topics);
        try {
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }


    /**
     * 获取当前主题的描述信息
     */
    @Test
    public void describeTopicConfig() {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        DescribeConfigsResult result =
                adminClient.describeConfigs(Collections.singleton(configResource));
        try {
            Config config = result.all().get().get(configResource);
            Collection<ConfigEntry> entries = config.entries();
            // 当前Topic的所有配置信息
            for (ConfigEntry entry : entries) {
                System.out.println(entry);
            }
            System.out.println("=====获取指定名称的配置信息======");
            ConfigEntry configEntry = config.get("cleanup.policy");
            System.out.println(configEntry);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 修改主题的配置参数
     */
    @Test
    public void alterTopicArg() {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        // 要修改的配置 k-v 的形式
        ConfigEntry configEntry = new ConfigEntry("cleanup.policy", "compact");
        AlterConfigOp alterConfigOp = new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET);
        Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
        configs.put(configResource, Collections.singleton(alterConfigOp));
        // 修改配置
        AlterConfigsResult result = adminClient.incrementalAlterConfigs(configs);
        try {
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 增加一个主题分区
     * 将 topic 的分区从4增加到5
     */
    @Test
    public void createPartitions() {
        NewPartitions newPartitions = NewPartitions.increaseTo(5);
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put(topic, newPartitions);
        CreatePartitionsResult result = adminClient.createPartitions(newPartitionsMap);
        try {
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

}
